package org.example.reducejoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * @ClassName TableMapper
 * @Author 小坏
 * @Date 2021/7/29、17:09
 * @Version 1.0
 * <p>
 * LongWritable: 偏移量,
 * Text：一行的内容
 */
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {

    private String fileName;

    private Text outK = new Text();
    private TableBean outV = new TableBean();

    /**
     * 初始化的时候想获取到文件的名称
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //初始化 order pd 只调用一次获取一个名称就够了
        FileSplit inputSplit = (FileSplit) context.getInputSplit(); //得到一个切片信息
        //ctrl+alt +f 提取全局
        fileName = inputSplit.getPath().getName(); //获取文件名称
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString(); //获取一行

        //判断是哪个文件
        if (fileName.contains("order")) {
            //处理订单表
            String[] split = line.split("\t");

            /**
             *  order
             * 1001	01	1
             * 1002	02	2
             * 1003	03	3
             * 1004	01	4
             * 1005	02	5
             * 1006	03	6
             */

            //封装K V
            outK.set(split[1]);
            outV.setId(split[0]);
            outV.setPid(split[1]);
            outV.setAmount(Integer.parseInt(split[2]));
            outV.setPname("");
            outV.setFlag("order");

        } else {
            //处理商品表

            /**
             *  bd 表
             *  01	小米
             *  02	华为
             *  03	格力
             */
            String[] split = line.split("\t");
            outK.set(split[0]);
            outV.setId("");
            outV.setPid(split[0]);
            outV.setAmount(0);
            outV.setPname(split[1]);
            outV.setFlag("pd");
        }

        //写出
        context.write(outK, outV);

    }
}
